package com.atguigu.pj.app

import java.text.DecimalFormat

import com.atguigu.pj.bean.UserVisitAction
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD

/**
 * description ：页面单跳转化率统计
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/1/10 20:34
 * modified By ：
 * version:    : 1.0
 */
object QueryPageSkipRate {


  def doQuery(sc: SparkContext, list: RDD[UserVisitAction], pages: String) = {

    // 处理页面跳转流 1,2,3,4,5 ==》 1->2, 2->3,3->4,4-5
    val pageList: Array[String] = pages.split(",")
    val beforeP: Array[String] = pageList.slice(0, pageList.length - 1) // 1 2 3 4
    val afterP: Array[String] = pageList.slice(1, pageList.length) // 2 3 4 5
    val skipList: Array[String] = beforeP.zip(afterP).map {
      case (before, after) => before + "->" + after
    }

    // 设置广播变量
    val pageListBC: Broadcast[Array[String]] = sc.broadcast(pageList)
    val skipListBC: Broadcast[Array[String]] = sc.broadcast(skipList)

    // 计算单个页面的点击次数
    val singlePageTotal: collection.Map[Long, Long] = list.filter(x => pageListBC.value.contains(x.page_id.toString))
      .map(x => (x.page_id, 1))
      .countByKey()


    // 计算页面跳转流数量
    val actionPageCount: collection.Map[String, Long] = list.groupBy(x => x.session_id) // sid, RDD[UserVisitAction]
      .sortBy {
        case (k, it) => if (null != it && null != it.head) it.head.action_time else k
      }
      .flatMap {
        case (_, it) =>
          val beforeU = it.slice(0, it.size - 1)
          val afterU = it.slice(1, it.size)
          beforeU.zip(afterU).map {
            case (b, f) => (b.page_id + "->" + f.page_id, 1)
          }
      }
      .filter(x => skipListBC.value.contains(x._1))
      .countByKey()

    // 计算跳转率
    val df = new DecimalFormat(".00%")
    val result = actionPageCount.map {
      case (action_skip, count) =>
        val in = action_skip.split("->")(0)
        val total = singlePageTotal.getOrElse(in.toLong, Long.MaxValue)
        (action_skip, df.format(count.toDouble / total.toDouble))
    }


    // 输出
    println(result.mkString(","))
  }

}
